package com.intct.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.func.LookupAsyncFunc;
import com.intct.ods.OdsManApp;
import com.intct.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @author gufg
 * @since 2025-08-25 09:46
 */
public class DwdVehicleApp {
    public static void main(String[] args) throws Exception {

        // 从命令中获取参数
        ParameterTool paraTools = ParameterTool.fromArgs(args);
        String mvnProjectPath = OdsManApp.class.getResource("/test.properties").getFile();

        String filePath = paraTools.get("filePath", mvnProjectPath);
        String topic = paraTools.get("kafka-topic-name", "ods_locus");

        // 从配置文件中获取配置参数
        ParameterTool propertiesFile = ParameterTool.fromPropertiesFile(filePath);

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);

        // 从kakfa ods
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(propertiesFile, "ods_vehicle", "dwd_vehicle_id");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dwd_vehicle_id");
            /*
            样例：
            {
              "op": "r",
              "name": "vehicle_distribution",
              "after": {
                "actionTime": 1753778164,
                "areaId": "110119",
                "driverId": "567",
                "carShelfNumber": "Y749265100D",
                "car": "307",
                "type": "0",
                "speed": 31
              },
              "ts_ms": 1753778164980
            }
             */

        // 数据清理
        SingleOutputStreamOperator<JSONObject> mapDS = sourceDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String jsonStr) throws Exception {

                // 从json string转为json object
                JSONObject jsonObj = JSON.parseObject(jsonStr);

                return jsonObj.getJSONObject("after");
            }
        });

        // 1 过滤掉不在线车辆
        SingleOutputStreamOperator<JSONObject> filterDS = mapDS.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObj) throws Exception {

                return "0".equals(jsonObj.getString("type"));
            }
        });

        // TODO 2 规范
        // TODO 2.1 重复数据
        // TODO 2.2 空值
        // TODO 2.3 脱敏
        // TODO 2.4 单位

        // 3 退维  [lookup 异步IO+缓存]
        SingleOutputStreamOperator<String> unorderedWait = AsyncDataStream.orderedWait(filterDS,
                new LookupAsyncFunc("intct83", "dim_area_info", "areaId"),10, TimeUnit.SECONDS, 1000);

        unorderedWait.print("unorderedWait>>>");
        //4 写在Kafka
        unorderedWait.sinkTo(KafkaUtil.getKafkaSink(propertiesFile, "dwd_vehicle"));

        // 启动作业
        env.execute();
    }
}
